package m1;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws Exception {
       /*
        标识一类Consumer的集合名称，

        这类Consumer通常消费一类消息，且消费逻辑一致。
        同一个Consumer Group下的各个实例将共同消费
        topic的消息，起到负载均衡的作用。

        消费进度以Consumer Group为粒度管理，不同
        Consumer Group之间消费进度彼此不受影响，
        即消息A被Consumer Group1消费过，也会再
        给Consumer Group2消费。

        注： RocketMQ要求同一个Consumer Group的
        消费者必须要拥有相同的注册信息，即必须要听一样
        的topic(并且tag也一样)。
         */

    //创建消费者对象
        DefaultMQPushConsumer c = new DefaultMQPushConsumer("cons-group1");
    //设置name server
    c.setNamesrvAddr("192.168.64.141:9876");
    //设置从哪里订阅消息
    // 标签: *--所有标签 TagA--只收TagA  TagA||TagB||TagC--收多个
    c.subscribe("Topic1", "TagA");
    //设置消息监听器处理消息
    //Concurrently--会启动多个线程处理消息
    c.setMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            for (MessageExt msg : list) {
                String s = new String(msg.getBody());
                System.out.println("收到:"+s);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            // return ConsumeConcurrentlyStatus.RECONSUME_LATER //稍后重新消费
        }
    });
    //启动
   c.start();
    }
}
